﻿#include "main.h"

#include <cstdio>
#include <cstdint>
#include <fcntl.h>
#include <cstdlib>
#include <unistd.h>
#include <thread>
#include <string>
#include <vector>
#include <sys/un.h>
#include <arpa/inet.h>
#include <signal.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <sys/socket.h>
#include <sys/timerfd.h>
#include <sys/signalfd.h>

#include <google/protobuf/message.h>
#include <boost/property_tree/ptree.hpp>
#include <boost/property_tree/ini_parser.hpp>
#include <rapidjson/document.h>

#include <cpptools/out.h>
#include <cpptools/time.hpp>
#include <cpptools/net/socket.h>

#include "PubHttpWsSrv.h"
#include "PubSockClient.h"
#include "ApiHandle.h"
#include "node.h"


Main::dbpool_t*                     Main::dbpool = nullptr; // 数据库连接池
Main::tpool_t*                      Main::tpool = nullptr; // 线程池
std::string                         Main::secret; // 加密密钥
uint32_t                            Main::timeout = 1800;
int                                 Main::epollFd = -1; // 主线程epoll句柄
int                                 Main::eventFd = -1; // 主线程eventfd句柄，用于任务线程通知主线程处理任务
uint32_t                            Main::beginTimesec = ::time(nullptr); // 程序启动时间
std::deque<Main::func_t>            Main::tasks; // 推送任务队列
std::mutex                          Main::mtx; // 推送任务回主线程锁

static pub::SockClient* _logClient = nullptr; // 日志模块连接管理
static constexpr const uint64_t kInfo = ct::Out::flagValue({ (uint64_t)out::kBaseInfo });
static constexpr const uint64_t kErr = ct::Out::flagValue({ (uint64_t)out::kBaseErr });


// 转换输出调试字符串为值
uint64_t swapOutFlagString(const char* ptr) {
    uint64_t flag = 0;

    if (!strcasecmp(ptr, "baseInfo")) { flag |= 1ULL << (uint64_t)out::kBaseInfo; }
    else if (!strcasecmp(ptr, "baseError")) { flag |= 1ULL << (uint64_t)out::kBaseErr; }
    else if (!strcasecmp(ptr, "connInfo")) { flag |= 1ULL << (uint64_t)out::kConnInfo; }
    else if (!strcasecmp(ptr, "connErr")) { flag |= 1ULL << (uint64_t)out::kConnErr; }
    else if (!strcasecmp(ptr, "connSend")) { flag |= 1ULL << (uint64_t)out::kConnSend; }
    else if (!strcasecmp(ptr, "connRecv")) { flag |= 1ULL << (uint64_t)out::kConnRecv; }
    else if (!strcasecmp(ptr, "logInfo")) { flag |= 1ULL << (uint64_t)out::kLogInfo; }
    else if (!strcasecmp(ptr, "logErr")) { flag |= 1ULL << (uint64_t)out::kLogErr; }
    else if (!strcasecmp(ptr, "logSend")) { flag |= 1ULL << (uint64_t)out::kLogSend; }
    else if (!strcasecmp(ptr, "logRecv")) { flag |= 1ULL << (uint64_t)out::kLogRecv; }
    else if (!strcasecmp(ptr, "queryInfo")) { flag |= 1ULL << (uint64_t)out::kQueryInfo; } 
    else if (!strcasecmp(ptr, "queryErr")) { flag |= 1ULL << (uint64_t)out::kQueryErr; } 
    else if (!strcasecmp(ptr, "querySend")) { flag |= 1ULL << (uint64_t)out::kQuerySend; } 
    else if (!strcasecmp(ptr, "queryRecv")) { flag |= 1ULL << (uint64_t)out::kQueryRecv; }
    else if (!strcasecmp(ptr, "all")) { flag |= UINT64_MAX; }

    return flag;
}

// 发送日志(上下线或操作日志)
void Main::sendLog(const google::protobuf::Message& msg, uint8_t logType) {
    std::uint32_t len = msg.ByteSizeLong();

    if (!len)
        return;

    unsigned char buf[len + pub::kHeadLen]; // 使用刚好的空间大小

    if (msg.SerializeWithCachedSizesToArray(buf + pub::kHeadLen)) { // 使用cached,避免再次求长度
        std::uint32_t* ptr32 = (std::uint32_t*)(buf + pub::head::kLen);
        buf[pub::head::kPre] = pub::kPrefix;
        buf[pub::head::kCode] = pub::code::kLog;
        buf[pub::head::kIdent] = logType;
        buf[pub::head::kSuffix] = 0;
        *ptr32 = htonl(len);

        outDebug("SendLog type:%hhu, len:%u.\n", logType, len + pub::kHeadLen);
        _logClient->sendData(buf, sizeof(buf), pub::SockClient::kDataAlways);
    }
}

// 新连接到来epoll回调
static void callbackNewConnection(Callback* cb, uint32_t event) {
    int fd, ret;
    struct sockaddr_storage addr;
    socklen_t socketLen = sizeof(addr);

    while ((fd = accept(cb->fd(), (struct sockaddr*)&addr, &socketLen)) > 0) {
        pub::HttpWebsocketServer::recv_buf_t buf(4096);
        ret = fcntl(fd, F_GETFL, 0); // 设置为非阻塞, 已经是延时accept
        ret |= O_NONBLOCK;
        fcntl(fd, F_SETFL, ret);

    #ifndef NDEBUG
        if (addr.ss_family == AF_INET) {
            struct sockaddr_in* addr_in = (struct sockaddr_in*)&addr;
            char ipstr[INET_ADDRSTRLEN];
            printf("New connection(%d), %s:%hu.\n", fd,
                   inet_ntop(AF_INET, &addr_in->sin_addr.s_addr, ipstr, sizeof(ipstr)),
                   ntohs(addr_in->sin_port));
        } else {
            struct sockaddr_un* addr_un = (struct sockaddr_un*)&addr;
            printf("New connection(%d), path: %s.\n", fd, addr_un->sun_path);
        }
    #endif

        // 读取并解析数据(非阻塞但延时accept，必定有数据了)
        if (buf.recv(fd) != session_t::recv_buf_t::AGAIN || buf.size() == 0) {
            outDebug("New connection(%d) recv failed, close it.\n", fd);
            close(fd);
            continue;
        }

         // http协议解析, 可能是普通http请求或websocket
        auto hw = new pub::HttpWebsocketServer(
            [] (pub::HttpWebsocketServer* conn, ct::Http* parser) {
                (new HttpHandle(conn, parser))->operator()();
            },
            [] (pub::HttpWebsocketServer* s, pub::HttpWebsocketServer::recv_buf_t&& buf) {
                auto uptr = std::make_unique<ApiHandle>(s, std::forward<pub::HttpWebsocketServer::recv_buf_t>(buf));
                uptr->operator()(std::move(uptr));
            },
            Main::epollFd,
            fd,
            std::move(buf)
        );

        hw->start();
    }
};

static void configInit(const char* filename) {
    namespace pt = boost::property_tree;
    pt::ptree config;
    std::string str;
    char buf[1024];

    pt::ini_parser::read_ini(filename, config); // 读取INI（可能抛出异常）

    // 调试相关
    str = config.get<std::string>("log-path", "");
    if (!str.empty()) {
        ct::Out::redirect(str.data(), STDOUT_FILENO);
    }

    str = config.get<std::string>("err-path", "");
    if (!str.empty()) {
        ct::Out::redirect(str.data(), STDERR_FILENO);
    }

    str = config.get<std::string>("log-flag", "0");
    uint64_t flag = ct::Out::parseFlag(str.data(), swapOutFlagString);
    ct::Out::setFlag(flag);

    // 数据库
    auto& db = config.get_child("db");
    auto name = db.get<std::string>("name", "dcp5");
    auto server = db.get<std::string>("server", "localhost");
    auto user = db.get<std::string>("user", "db-user");
    auto password = db.get<std::string>("password", "db-password");

    ct::Out::flag(kInfo, "Database --> host(%s), user(%s), password(%s), name(%s).\n",
                  server.data(), user.data(), password.data(), name.data());
    Main::dbpool = new Main::dbpool_t(name.data(), server.data(), user.data(), password.data(),
                                      2, 8, true, Main::dbpool_t::CharSet::UTF8MB4);

    // 日志模块
    auto& module = config.get_child("module");
    str = module.get<std::string>("log");
    int type;
    auto addr = pub::SockClient::parseUri(str.data(), &type);
    if (!addr) {
        fprintf(stderr, "Error --> log address parse failed: %s.\n", str.data());
        exit(EXIT_FAILURE);
    }
    ct::Out::flag(kInfo, "log --> %s.\n", str.data());
    _logClient = new pub::SockClient((uint16_t)1, 3, 8, addr, Main::epollFd, []{}, [](pub::Pack&){});

    // 绑定信息
    bool service = false;
    auto& bind = config.get_child("bind");
    auto tcp = bind.get<std::string>("tcp", "0.0.0.0:10005");
    sprintf(buf, "tcp://%s", tcp.data());

    int fd = ct::net::SocketUtil::server(buf, ct::net::SocketUtil::reuseAddr + ct::net::SocketUtil::reusePort);

    if (fd < 0) {
        ct::Out::flag(kErr, "Error --> Service create failed: %s\n", strerror(errno));
        exit(EXIT_FAILURE);
    }

    ct::Out::flag(kInfo, "Service --> %s.\n", buf);
    ct::net::SocketUtil::setDeferaccept(fd, 5); // 设置延时accept
    int ret = fcntl(fd, F_GETFL, 0); // 设置为非阻塞
    ret |= O_NONBLOCK;
    fcntl(fd, F_SETFL, ret);

    struct epoll_event event; // 加入epoll监控
    Callback* cb = new Callback(fd);
    cb->func = callbackNewConnection;
    event.events = EPOLLIN | EPOLLET; // ET模式
    event.data.ptr = cb;
    epoll_ctl(Main::epollFd, EPOLL_CTL_ADD, fd, &event);

    service = true;

    // 本地UNIX域
    auto local = bind.get<std::string>("local", "");

    if (!local.empty()) {
        unlink(local.data()); // 删除旧文件
        sprintf(buf, "lsp://%s", local.data());
        fd = ct::net::SocketUtil::server(buf, ct::net::SocketUtil::reuseAddr); // 创建socket

        if (fd < 0) {
            ct::Out::flag(kErr, "Error --> Service create failed: %s\n", strerror(errno));
            exit(EXIT_FAILURE);
        }

        ct::Out::flag(kInfo, "Service --> %s.\n", buf);
        ct::net::SocketUtil::setDeferaccept(fd, 5); // 设置延时accept
        ret = fcntl(fd, F_GETFL, 0); // 设置为非阻塞
        ret |= O_NONBLOCK;
        fcntl(fd, F_SETFL, ret);

        struct epoll_event event; // 加入epoll监控
        Callback* cb = new Callback(fd);
        cb->func = callbackNewConnection;
        event.events = EPOLLIN | EPOLLET; // ET模式
        event.data.ptr = cb;
        epoll_ctl(Main::epollFd, EPOLL_CTL_ADD, fd, &event);
        service = true;
    }

    if (!service) { // 网络和本地绑定必须有一个
        ct::Out::flag(kErr, "Error --> No Service.\n");
        exit(EXIT_FAILURE);
    }
}

#if 0
// 信号捕获初始化
void signalInit(void) {
    sigset_t mask;
    sigemptyset(&mask);

    // 需要捕获处理的退出信号
    sigaddset(&mask, SIGINT);
    sigaddset(&mask, SIGTERM);
    sigaddset(&mask, SIGQUIT);

    // 忽略的信号
    struct sigaction sa_ignore {}; // 所有字段0
    sa_ignore.sa_handler = SIG_IGN;
    sigaction(SIGHUP, &sa_ignore, NULL);
    sigaction(SIGPIPE, &sa_ignore, NULL);
    sigprocmask(SIG_BLOCK, &mask, NULL); // 阻塞目标信号，防止默认处理

    int fd = signalfd(-1, &mask, 0);

    if (fd < 0) {
        ct::Out::flag(kErr, "Error --> Signalfd create failed: %s.\n", strerror(errno));
        exit(EXIT_FAILURE);
    }

    struct epoll_event event;
    Callback* cb = new Callback(fd); // 加入epoll监控
    cb->func = [](Callback* cb, uint32_t event) { // 回调
        
    };

    event.events = EPOLLIN;
    event.data.ptr = cb;
    epoll_ctl(Main::epollFd, EPOLL_CTL_ADD, fd, &event);
}
#endif

// 定时器初始化
static void timerInit(void) {
    struct itimerspec timerspec;
    struct epoll_event event;
    Callback* cb;

    int fd = timerfd_create(CLOCK_MONOTONIC, 0);
    timerspec.it_interval.tv_nsec = 0;
    timerspec.it_interval.tv_sec = 3600; // 首次超时之后每次超时时间，0表示禁用
    timerspec.it_value.tv_nsec = 0;
    timerspec.it_value.tv_sec = 3600; // 首次超时, 距离下一小时间隔
    timerfd_settime(fd, 0, &timerspec, NULL);

    cb = new Callback(fd); // 加入epoll监控
    cb->func = [](Callback* cb, uint32_t event) { // 回调
        uint64_t val; // 读掉, 否则下次频繁通知
        val = read(cb->fd(), &val, sizeof(val));

        uint32_t timesec = (uint32_t)time(nullptr);
        uint32_t timeoutsec;
        char buf[40];
        //ct::Out::debug("timeout, current time: %s.\n", ct::Time::timestampToString((time_t)timesec, buf));

        for (auto it = Token::all.begin(); it != Token::all.end(); ) {
            auto& tokenSptr = it->second;
            timeoutsec = tokenSptr->lastTime() + tokenSptr->timeout();
            /*auto& accSptr = tokenSptr->accountSptr();
            printf("Account(%s), ip(%s), ontime(%s), lasttime(%s).\n",
                   accSptr->name().data(), tokenSptr->ip().data(),
                   ct::Time::timestampToString((time_t)tokenSptr->onTime(), buf),
                   ct::Time::timestampToString((time_t)tokenSptr->lastTime(), buf + 20)
            );*/
            if (timeoutsec < timesec) { // 超时
                ct::Out::flag(kInfo, "Account(%s) token(%lu) timeout, ontime:%s.\n",
                              tokenSptr->accountSptr()->name().data(), tokenSptr->token(),
                              ct::Time::timestampToString((time_t)it->second->upTime(), buf));
                tokenSptr->logoutCaugh(pub::off::kTimeout);
                tokenSptr->lastTime(timeoutsec); // 置下线时间
                Token::all.erase(it++);
            } else {
                it++;
            }
        }
    };

    event.events = EPOLLIN;
    event.data.ptr = cb;
    epoll_ctl(Main::epollFd, EPOLL_CTL_ADD, fd, &event);
}

static void eventInit(void) {
    Main::eventFd = eventfd(0, EFD_CLOEXEC);

    if (Main::eventFd < 0) {
        ct::Out::error("Error --> Eventfd create failed: %s.\n", strerror(errno));
        exit(EXIT_FAILURE);
    }

    struct epoll_event event;
    Callback* cb;

    cb = new Callback(); // 加入epoll监控
    cb->func = [](Callback* cb, uint32_t event) { // 回调
        uint64_t val; // 读掉
        val = read(Main::eventFd, &val, sizeof(val));
        Main::mtx.lock();
        auto tasks(std::move(Main::tasks));
        Main::mtx.unlock();
        for (auto& itr : tasks) { // 逐个执行推送的函数
            itr();
        }
    };

    event.events = EPOLLIN;
    event.data.ptr = cb;
    epoll_ctl(Main::epollFd, EPOLL_CTL_ADD, Main::eventFd, &event);
}

int main(int argc, char* argv[]) {
    if (argc < 2) {
        fprintf(stderr, "Usage: %s <config-file>\n", argv[0]);
        exit(EXIT_FAILURE);
    }

    srand(::time(nullptr));
    Main::epollFd = epoll_create(10);
    configInit(argv[1]); // 读取配置文件
    timerInit(); // 定时器初始化
    eventInit(); // 事件通知初始化
    Main::tpool = new Main::tpool_t(1, 4, true); // 线程池
    Permission::loadData();
    Role::loadData();
    _logClient->connecting(); // 开始连接到日志模块

    int i;
    struct epoll_event events[4];

    while ((argc = epoll_wait(Main::epollFd, events, 4, -1)) > 0) {
        for (i = 0; i < argc; ++i) {
            ((Callback*)events[i].data.ptr)->func((Callback*)events[i].data.ptr, events[i].events); // 等同于下面两句
        }
    }

    return 0;
}